package com.hellorxjava.demo;

import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.core.Scheduler;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.schedulers.Schedulers;

/**
 * Scheduler分  Schedulers.SINGLE  Schedulers.COMPUTATION  Schedulers.IO Schedulers.TRAMPOLINE Schedulers.NEW_THREAD
 * computation适用于和cpu有关的任务 但不适合那些造成阻塞的任务
 *  cpu核心数来创建线程池 每个线程占用一个cpu核心可以充分利用cpu的计算资源
 * newThread 会每次新建一个线程 一般情况下不推荐使用这个Scheduler  适合使用于工作时间长并且总数较少的任务
 * io的scheduler可以回收利用 内部也维护了一个线程池 开始查找空闲的线程池 有执行任务 没有就创建线程并且加这个线程加入线程池 超过60s后就被回收
 * 因为io scheduler不限制大小 所以可能会造成内存溢
 *
 * TRAMPOLINE 表示在当前线程执行任务 线程的任务执行完成了之后再执行任务
 * from(Executor e) 根据我们提供的Executor创建Scheduler
 */
public class RXJavaDemo7 {
    public static void main(String[] args) {
        Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> emitter) throws Throwable {
                System.out.println("start" + Thread.currentThread().getName());
                emitter.onNext(1);
                emitter.onComplete();

            }
        }).subscribeOn(Schedulers.newThread())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Throwable {
                        System.out.println(integer + "start" + Thread.currentThread().getName());
                        return integer + 1;
                    }
                })
                .observeOn(Schedulers.io())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Throwable {
                        System.out.println(integer + "start" + Thread.currentThread().getName());
                        return integer + 1;
                    }
                }).subscribeOn(Schedulers.computation())
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {

                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        System.out.println(integer + "onNext" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });

        ThreadPoolExecutor executor = new ThreadPoolExecutor(2, 2, 2000L, TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<Runnable>(1000),
                new SimpleThreadFactory()
        );

        @NonNull Scheduler scheduler = Schedulers.from(executor);
        Observable.just(1,2,3,4,5,6).take(5).observeOn(scheduler)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        System.out.println("scheduler onSubscribe");
                    }

                    @Override
                    public void onNext(@NonNull Integer integer) {
                        System.out.println(integer + " onNext " + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }


    static class SimpleThreadFactory implements ThreadFactory {

        @Override
        public Thread newThread(Runnable runnable) {
            return new Thread(runnable);
        }
    }
}
